package com.taobao.txc.sample;

import com.aliyun.openservices.ons.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.atomic.AtomicInteger;

@Service
public class BizServiceConsumer implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(BizServiceConsumer.class);

    private static final String TOPIC = "gts-test-topic";

    @Autowired
    private Consumer consumer;

    @Override
    public void afterPropertiesSet() throws Exception {
        final AtomicInteger transferCount = new AtomicInteger(0);
        final AtomicInteger transferMoneyCount = new AtomicInteger(0);
        consumer.subscribe(TOPIC, "*", new MessageListener() {
            public Action consume(Message message, ConsumeContext context) {
                String v = new String(message.getBody());
                int money = Integer.parseInt(v);
                transferCount.incrementAndGet();
                transferMoneyCount.addAndGet(money);
                LOGGER.info(">>> Check Receive Message " + transferCount + " transactions was done with " + transferMoneyCount);
                return Action.CommitMessage;
            }
        });
        consumer.start();
        LOGGER.info("MQ Consumer is started.");

    }
}
